package com.catmiao.spark.stream

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @title: SparkStreaming01_WordCount
 * @projectName spark_study
 * @description: TODO
 * @author ChengMiao
 * @date 2024/3/25 00:31
 */
object SparkStreaming05_State {

  def main(args: Array[String]): Unit = {


    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    // 需要设定缓冲存储的路径 缓冲点
    ssc.checkpoint("cp")

    // 无状态数据操作，只对当前采集周期内的数据进行处理
    // 在某些场合下，需要保留数据统计的结果【状态】，实现数据的汇总
    val socketInputStream: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    val mapToOne: DStream[(String, Int)] = socketInputStream.map((_, 1))

//    val result: DStream[(String, Int)] = mapToOne.reduceByKey(_ + _)
//    result.print()


    /**
     * updateStateByKey 根据key对数据的状态进行更新
     *  - param1：相同key的value数据
     *  - param2：缓冲区相同key的value数据
     *
     * 使用有状态操作时，需要设定检查点路径 checkPoint
     */
    val state: DStream[(String, Int)] = mapToOne.updateStateByKey(
      (seq: Seq[Int], opt: Option[Int]) => {
        // 累加值
        val newCount = opt.getOrElse(0) + seq.sum
        // 放入缓冲区
        Option(newCount)
      }
    )

    state.print()

    ssc.start()
    ssc.awaitTermination()
  }


}
